package com.cy.ragbase.consumer;

import com.cy.ragbase.dto.DocumentProcessingTask;
import com.cy.ragbase.service.DocumentService;
import com.cy.ragbase.service.RedisQueueService;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicBoolean;

@Component
@RequiredArgsConstructor
@Slf4j
public class DocumentProcessingConsumer implements CommandLineRunner {

    private final RedisQueueService redisQueueService;
    private final DocumentService documentService;
    private final AtomicBoolean running = new AtomicBoolean(true);

    @Override
    @Async
    public void run(String... args) {

        while (running.get()) {
            try {
                // 从队列获取任务，超时时间为5秒
                DocumentProcessingTask task = redisQueueService.dequeueDocumentProcessing(5);

                if (task != null) {
                    processTask(task);
                }


            } catch (Exception e) {
                log.error("文档处理消费者出错", e);
                // 短暂休眠后继续
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }

        log.info("文档处理消费者已停止");
    }
    // 处理文档
    /**
     * 处理单个任务
     */
    private void processTask(DocumentProcessingTask task) {
        try {
            // 处理文档
            documentService.processDocument(task);

            // 标记任务完成
            redisQueueService.markTaskCompleted(task);


        } catch (Exception e) {
            log.error("任务处理失败: {}", task.getTaskId(), e);

            // 标记任务失败，可能触发重试
            redisQueueService.markTaskFailed(task.getTaskId(), e.getMessage(), task);
        }
    }

    /**
     * 定期处理延迟重试任务
     */
    @Scheduled(fixedDelay = 30000) // 每30秒执行一次
    public void processDelayedTasks() {
        if (running.get()) {
            redisQueueService.processDelayedTasks();
        }
    }

    @PreDestroy
    public void shutdown() {
        log.info("关闭文档处理消费者...");
        running.set(false);
    }
}